/*
 * Licensed to the Tai Ping Jin Ke
 *
 * Copyright (c) 2022 .
 * All rights reserved.
 * 项目名称：太平金科-基础组件封装层-会话管理-核心依赖
 * 版权说明：本软件属太平金融科技服务(上海)有限公司所有，在未获得太平金融科技服务(上海)有限公司正式授权情况下，任何企业和个人，不能获取、阅读、安装、传播本软件涉及的任何受知识产权保护的内容。
 */
package com.taipingframework.boot.session;

import com.google.common.collect.Lists;
import com.taipingframework.boot.cache.core.ReactiveCacheHandler;
import com.taipingframework.boot.cache.redis.DynamicCacheKeyBuilder;
import com.taipingframework.boot.cache.redis.StandardCacheKeyBuilder;
import com.taipingframework.boot.session.constant.SessionCacheKeyEnum;
import com.taipingframework.boot.session.constant.SessionCachePathEnum;
import com.taipingframework.boot.session.constant.SessionConstant;
import com.taipingframework.boot.session.support.ClientDetails;
import com.taipingframework.boot.web.client.BrowserCookie;
import com.taipingframework.boot.web.client.ReactiveCookieScene;
import com.taipingframework.boot.web.pojo.KeyValuePair;
import com.taipingframework.boot.web.response.status.ApiStatusEnum;
import com.taipingframework.boot.web.response.status.ExceptionStatusEnum;
import com.taipingframework.utility.constant.ApplicationConstant;
import com.taipingframework.utility.constant.SystemConstant;
import com.taipingframework.utility.exception.SystemInterrupterException;
import com.taipingframework.utility.exception.UtilityException;
import com.taipingframework.utility.extend.ReactorHelper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.lang.NonNull;
import reactor.core.publisher.Mono;

import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

/**
 * WebSession会话控制
 */
@Slf4j
public class ReactorSessionManager {

    /**
     * 标准缓存(读/写)处理器
     */
    private static ReactiveCacheHandler cacheHandler;

    public ReactorSessionManager(ReactiveCacheHandler reactiveCacheHandler) {
        ReactorSessionManager.cacheHandler = reactiveCacheHandler;
        // 在每次应用启动时，刷新一次缓存中的数据，防止配置变更后无法及时读取到正确的配置信息
        reactiveCacheHandler.setValue(SessionCacheKeyEnum.SERVER_SESSION_PATH_INFO,
                DynamicCacheKeyBuilder.createUniqueCacheKey(SessionConstant.PLACEHOLDER_SESSION_ID, SessionCachePathEnum.SESSION_DATA_PATH))
                .onErrorResume(e -> Mono.error(new UtilityException(e)))
                .subscribe();
    }

    /**
     * 刷新当前账号在缓存服务器上面的会话数据
     *
     * @param loginName       登录名
     * @param clientIpAddress 终端显示层所在设备的IP
     */
    public static Mono<Void> refreshTerminalSession(String loginName, String clientIpAddress, String flagManagedProfile) {
        String uniqueTokenKey = buildUniqueTokenKey(loginName, clientIpAddress, flagManagedProfile);
        return Mono.when(
                // 刷新当前会话对象数据
                cacheHandler.expireByDeduce(uniqueTokenKey, SessionCachePathEnum.SESSION_UNEXPIRED_PATH),
                // 刷新当前账号在缓存服务器上面的终端用户空间列表数据
                cacheHandler.expireByDeduce(clientIpAddress, SessionCachePathEnum.SESSION_PROFILE_PATH),
                // 刷新当前账号在缓存服务器上面的会话对象列表数据
                cacheHandler.expireByDeduce(loginName, SessionCachePathEnum.SESSION_CREDENTIALS_PATH)
        ).onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 移除缓存对象在缓存服务器上面失效的缓存数据，并返回有效的缓存数据
     * <p>
     * 如果登录账号允许在多个终端同时登录，则该账号在缓存服务器上面会存在多个会话对象。
     * 另外，同一个账号在多个不同终端同时登录后，创建的多个会话对象之间，因为Spring-Session的会话过期机制在多个会话对象间是彼此隔离的，
     * 所以，它们在缓存服务器上面的会话数据也是必须进行逻辑隔离的。
     *
     * @param dynamicCacheKey 登录名或者终端Ip
     * @param cachePathEnum   缓存对象的路径：client-credentials对应的入参`dynamicCacheKey`应为登录名；client-profile对应的入参`dynamicCacheKey`应为终端IP
     * @return cachePathEnum参数为client-credentials对应返回的是Map<会话对象唯一标识, 终端ip>；cachePathEnum参数为client-profile对应返回的是Map<终端用户空间标识, 用户名>
     */
    public static Mono<Map<String, String>> removeExpiredTerminalSession(@NonNull String dynamicCacheKey, @NonNull SessionCachePathEnum cachePathEnum) {
        // 获取当前会话对象存放在缓存服务器的缓存数据
        return cacheHandler.getHash(dynamicCacheKey, cachePathEnum)
                .zipWhen(hashMap -> {
                            // 搜集登录过期的缓存数据
                            return Mono.just(hashMap)
                                    .flatMapIterable(Map::entrySet)
                                    .filterWhen(entry -> {
                                        // 此处取反，拿到的是过期缓存数据
                                        if (cachePathEnum == SessionCachePathEnum.SESSION_CREDENTIALS_PATH) {
                                            // 筛选过期的缓存数据
                                            return cacheHandler.hasKey(entry.getKey(), SessionCachePathEnum.SESSION_DATA_PATH)
                                                    .map(flag -> !flag);
                                        } else if (cachePathEnum == SessionCachePathEnum.SESSION_PROFILE_PATH) {
                                            String uniqueTokenKey = buildUniqueTokenKey(entry.getValue(), dynamicCacheKey, entry.getKey());
                                            // 筛选过期的缓存数据
                                            return cacheHandler.hasKey(uniqueTokenKey, SessionCachePathEnum.SESSION_UNEXPIRED_PATH)
                                                    .map(flag -> !flag);
                                        } else {
                                            // 在后续维护过程中，一定要明确设计思路，入参错误此处必抛异常
                                            throw new SystemInterrupterException(ApiStatusEnum.ERR_A0410);
                                        }
                                    })
                                    // 此处得到的 Flux<Map.Entry<String, String>> 是一个过期的缓存数据序列，使用 Java Stream 还原为 Map
                                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue, (oldKey, newKey) -> newKey, LinkedHashMap::new))
                                    // 返回过期的hashKey列表：client-credentials对应返回的是会话对象唯一标识列表；client-profile对应返回的是终端用户空间标识列表
                                    .map(expiredHashMap -> Lists.newArrayList(expiredHashMap.keySet()));
                        },
                        (hashMap, expiredHashKeyList) -> {
                            // 移除当前账号在缓存服务器上面过期的缓存数据
                            if (expiredHashKeyList.size() > 0) {
                                for (String hashKey : expiredHashKeyList) {
                                    hashMap.remove(hashKey);
                                }
                                return cacheHandler.removeHashEntry(dynamicCacheKey, cachePathEnum, expiredHashKeyList)
                                        .thenReturn(hashMap);
                            }
                            return Mono.just(hashMap);
                        })
                .flatMap(credentialsMapMono -> credentialsMapMono)
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 在缓存中添加/更新验证信息
     *
     * @param clientDetails 客户端发送HTTP请求的参数详细
     * @param sessionId     会话对象标识
     */
    public static Mono<Void> saveOrUpdateCurrentTerminalSession(ClientDetails clientDetails, String sessionId) {
        String loginName = clientDetails.getClientLoginName();
        String clientIpAddress = clientDetails.getClientIpAddress();
        String flagManagedProfile = clientDetails.getClientUserProfile();
        String uniqueTokenKey = buildUniqueTokenKey(loginName, clientIpAddress, flagManagedProfile);

        return Mono.when(
                // 将当前会话对象标识记录在缓存服务器，用于说明当前会话对象是有效的
                cacheHandler.setValue(uniqueTokenKey, SessionCachePathEnum.SESSION_UNEXPIRED_PATH, sessionId),
                // 在缓存服务器记录当前账号在终端的用户空间信息
                cacheHandler.putHash(clientIpAddress, SessionCachePathEnum.SESSION_PROFILE_PATH, flagManagedProfile, loginName),
                // 在缓存服务器记录当前账号登录成功后创建的会话对象标识
                cacheHandler.putHash(loginName, SessionCachePathEnum.SESSION_CREDENTIALS_PATH, sessionId, clientIpAddress)
        ).onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 删除验证信息
     */
    public static Mono<Void> deleteCurrentTerminalSession(String loginName, String clientIpAddress, String flagManagedProfile, String sessionId) {
        return Mono.when(
                cacheHandler.delete(buildUniqueTokenKey(loginName, clientIpAddress, flagManagedProfile), SessionCachePathEnum.SESSION_UNEXPIRED_PATH),
                cacheHandler.removeHashEntry(clientIpAddress, SessionCachePathEnum.SESSION_PROFILE_PATH, Collections.singletonList(flagManagedProfile)),
                cacheHandler.removeHashEntry(loginName, SessionCachePathEnum.SESSION_CREDENTIALS_PATH, Collections.singletonList(sessionId))
        ).onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 读取当前账号通过当前终端设备在缓存服务器创建的会话对象标识
     */
    public static Mono<String> readCurrentTerminalSession(String loginName, String clientIpAddress, String flagManagedProfile) {
        return cacheHandler.getValue(buildUniqueTokenKey(loginName, clientIpAddress, flagManagedProfile), SessionCachePathEnum.SESSION_UNEXPIRED_PATH)
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 校验当前登录名在缓存服务器是否已经存在会话对象数据
     * <p>
     * PS：虽然登录接口被权限验证模块加入白名单，这样使得登录接口在被调用时不需要提供Access-Token，但是如果进行严谨的逻辑验证：
     * 在会话对象的有效期内，一个账户的会话对象必须是唯一的，它在客户端的凭证也必须是唯一的，既不能被租借也防止被盗取！
     * <p>
     * 实际上，根据Spring-Session框架的特点，无论是在请求头中还是在Cookie中，每次HTTP请求只要可以拿到一个有效的Access-Token，
     * 那么当前HTTP请求处理线程肯定已经被绑定在了某个唯一的会话对象之下的，因此，保护会话对象的安全，就是保护业务数据的安全！
     * <p>
     * 另外，在基于传统Servlet原生Session框架中，由于会话对象在分布式集群架构下都是独立的，想要实现会话对象的唯一性，一般都会通过
     * 会话共享、会话复制、会话保持三种方式达到目的，但性能不佳，也跟不上互联网技术发展的步伐，基本已经不用。
     *
     * @param clientDetails 客户端发送HTTP请求的参数详细
     * @return true - 表示需要读取当前会话对象并返回；false - 表示需要创建会话对象并返回
     */
    public static Mono<Boolean> validateClientInfo(ClientDetails clientDetails) {
        return Mono.defer(() -> Mono.fromSupplier(() -> ClientDetails.validateForSession(clientDetails)))
                .filter(StringUtils::isNotEmpty)
                .hasElement()
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 读取客户端发送HTTP请求的参数详细
     */
    public static Mono<ClientDetails> readClientDetails(String loginName) {
        return ReactorHelper.getExchange()
                .flatMap(exchange -> {
                    // 当前HTTP请求的唯一标识
                    final String clientRequestId = Optional.ofNullable((String) exchange.getAttribute(ApplicationConstant.RESPONSE_IDENTITY))
                            .orElse(StringUtils.EMPTY);
                    // 获取HTTP请求的远程客户端设备标识
                    String clientIpAddress = Optional.ofNullable((String) exchange.getAttribute(ApplicationConstant.REMOTE_CLIENT_IP_KEY))
                            .orElse(StringUtils.EMPTY);
                    // 获取请求头中的token
                    String requestHeaderToken = Optional.ofNullable(exchange.getRequest().getHeaders().getFirst(StandardCacheKeyBuilder.getAuthTokenKey()))
                            .orElse(StringUtils.EMPTY);

                    // 装载会话管理模块功能支撑详细参数
                    return Mono.just(new ClientDetails())
                            .flatMap(clientDetails -> {
                                clientDetails.setClientRequestId(clientRequestId);
                                clientDetails.setClientIpAddress(clientIpAddress);
                                clientDetails.setRequestHeaderToken(requestHeaderToken);
                                clientDetails.setClientLoginName(loginName);
                                return Mono.just(clientDetails);
                            })
                            .zipWith(readClientUserProfile().zipWhen(clientUserProfile ->
                                            cacheHandler.getValue(buildUniqueTokenKey(loginName, clientIpAddress, clientUserProfile),
                                                    SessionCachePathEnum.SESSION_UNEXPIRED_PATH).defaultIfEmpty(StringUtils.EMPTY), KeyValuePair::new),
                                    (clientDetails, keyValuePair) -> {
                                        clientDetails.setClientUserProfile(keyValuePair.getKey());
                                        clientDetails.setClientPresentToken((String) keyValuePair.getValue());
                                        return clientDetails;
                                    })
                            .zipWith(removeExpiredTerminalSession(clientIpAddress, SessionCachePathEnum.SESSION_PROFILE_PATH),
                                    (clientDetails, serverUserProfileMap) -> {
                                        clientDetails.setServerUserProfileMap(serverUserProfileMap);
                                        return clientDetails;
                                    })
                            .zipWith(removeExpiredTerminalSession(loginName, SessionCachePathEnum.SESSION_CREDENTIALS_PATH),
                                    (clientDetails, serverCredentialsMap) -> {
                                        clientDetails.setServerCredentialsMap(serverCredentialsMap);
                                        return clientDetails;
                                    })
                            .zipWith(exchange.getSession(),
                                    (clientDetails, webSession) -> {
                                        if (webSession.isStarted() && webSession.getId().equals(requestHeaderToken)) {
                                            clientDetails.setSessionPrincipal(webSession.getAttributeOrDefault(SessionConstant.SESSION_LOGIN_KEY, StringUtils.EMPTY));
                                            clientDetails.setSessionClientIpAddress(webSession.getAttributeOrDefault(ApplicationConstant.REMOTE_CLIENT_IP_KEY, StringUtils.EMPTY));
                                            clientDetails.setSessionUserProfile(webSession.getAttributeOrDefault(SessionConstant.CLIENT_USER_PROFILE_KEY, StringUtils.EMPTY));
                                        }
                                        return clientDetails;
                                    }
                            );
                })
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

    /**
     * 构建HTTP会话对象的属性名
     */
    @NonNull
    public static String buildAttributeCacheKey(String hashName) {
        return String.join(SystemConstant.CACHE_SEPARATOR, SessionConstant.PREFIX_SESSION_ATTRIBUTE, hashName);
    }

    /**
     * 用户会话标识
     */
    public static String buildUniqueTokenKey(String loginName, String clientIpAddress, String clientUserProfile) {
        return String.join(SystemConstant.TITLE_SEPARATOR, clientIpAddress, loginName, clientUserProfile);
    }

    /**
     * 读取客户端发送HTTP请求的用户空间
     */
    public static Mono<String> readClientUserProfile() {
        return ReactorHelper.getExchange()
                .map(exchange -> Optional.ofNullable(exchange.getRequest().getHeaders().getFirst(SessionConstant.MANAGED_PROFILE_KEY))
                        .orElse(StringUtils.EMPTY))
                .filter(StringUtils::isNotEmpty)
                .switchIfEmpty(ReactiveCookieScene.readWebCookie(SessionConstant.MANAGED_PROFILE_KEY).map(BrowserCookie::getValue))
                .map(flagManagedProfile -> {
                    if (StringUtils.isEmpty(flagManagedProfile) || StringUtils.isBlank(flagManagedProfile)) {
                        throw new SystemInterrupterException(ExceptionStatusEnum.UNABLE_GET_MANAGED_PROFILE);
                    }
                    return flagManagedProfile;
                })
                .onErrorResume(e -> Mono.error(new UtilityException(e)));
    }

}
